package sc.analysis.ppo;


import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import sc.writable.IntermediateData;
import sc.writable.IntermediateData.Type;


/**
 * Reducer of the Percentage Price Oscillator (PPO)
 *
 * @author Quentin Rossé
 *
 */
public class PpoReducer extends Reducer<Text, PpoData, 
                                        Text, IntermediateData> {
	
	private static final String ANALYSIS_NAME = "PPO (MAC-D)";
	
	
	/**
	 * data written for the next phase
	 */
	private static IntermediateData imd = new IntermediateData();
	
	@Override
	public void reduce(Text key, Iterable<PpoData> values, Context context) throws IOException, InterruptedException {
		System.out.println(key);
		String details;
		
		for(PpoData ppod : values){
			System.out.println(//"ppo : " + ppod.getPpo() + "  " + 
					            //ppod.getPrevPpo() + 
					            "   histo : " + ppod.getHistogram()); //+ "  " +
					            //ppod.getPrevHistogram());
			
			//zero line cross
			if(ppod.getPrevPpo() < 0 && ppod.getPpo() >= 0){
				imd.setScore(1);
				imd.setType(Type.ALERT_BULL);
				imd.setDetails("PPO crossed the zero line");
				context.write(key, imd);
			}
			else if(ppod.getPrevPpo() > 0 && ppod.getPpo() <= 0){
				imd.setScore(1);
				imd.setType(Type.ALERT_BEAR);
				imd.setDetails("PPO crossed the zero line");
				context.write(key, imd);
			}
			
			//signal and ppo cross
			if(ppod.getPrevHistogram() < 0 && ppod.getHistogram() >= 0){
				imd.setScore(1);
				imd.setType(Type.ALERT_BULL);
				imd.setDetails("PPO crossed the signal line");
				context.write(key, imd);
			}
			else if(ppod.getPrevHistogram() > 0 && ppod.getHistogram() <= 0){
				imd.setScore(1);
				imd.setType(Type.ALERT_BEAR);
				imd.setDetails("PPO crossed the signal line");
				context.write(key, imd);
			}
			
			//general score
			imd.setScore(score(ppod.getPpo(), ppod.getHistogram()));
			imd.setType(Type.SHORT_TERM_INDICATOR);
			details = "";
			if(ppod.getPpo() >= 0f)
				details += "Above zero line ";
			else
				details += "Below zero line ";
			if(ppod.getHistogram() >= 0f)
				details += "PPO above signal";
			else
				details += "PPO below signal";
			imd.setDetails(details);
			context.write(key, imd);
		}
	}
	
	@Override
	public void setup(Context context){
		imd.setAnalysisName(ANALYSIS_NAME);
	}
	
	@Override
	public void cleanup(Context context){}
	
	
	/**
	 * calculate the score from the ppo and the value of the histogram
	 * 
	 * @param ppo value of the ppo
	 * @param histogram value of the histogram (ppo - signal)
	 * 
	 * @return the score
	 */
	private int score(float ppo, float histogram){
		int score;
		
		//ppo above zero line or not
		if(ppo >= 0)
			score = 6;
		else
			score = 4;
		
		//big move in histogram or not
		if(Math.abs(histogram) > 1.5f)
			score += (histogram > 0 ? 2 : -2);
		else if(Math.abs(histogram) > 0.5f)
			score += (histogram > 0 ? 1 : -1);
		
		return score;
	}
}